package org.example.aggres;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * rich function
 */
public class DataRichFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Dog> source = env.fromElements(
                new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1),
                new Dog("bb", "b", 5), new Dog("bb", "b", 5), new Dog("bb", "b", 5),
                new Dog("cc", "b", 6), new Dog("cc", "b", 6),
                new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8),
                new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4),
                new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4)
        );
        env.setParallelism(5);
        source.map(new RichMapFunction<Dog, Dog>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("open---" + getRuntimeContext().getIndexOfThisSubtask());
                super.open(parameters);
            }

            @Override
            public void close() throws Exception {
                System.out.println("close---" + getRuntimeContext().getIndexOfThisSubtask());
                super.close();
            }

            @Override
            public RuntimeContext getRuntimeContext() {
                return super.getRuntimeContext();
            }

            @Override
            public Dog map(Dog value) throws Exception {
                return value;
            }
        }).print();

        env.execute();
    }
}